package org.pqw.bookmodel.config;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;
import org.pqw.bookmodel.config.enums.MqTagEnum;
import org.pqw.bookmodel.config.enums.MqTopicEnum;

import javax.annotation.Resource;
import java.util.List;

/**
 * 抽象消费者
 *
 * @author 白衣雪
 * @since 2021/6/21 22:34
 */
public abstract class AbstractMqConsumer {

    @Resource
    private DefaultMQPushConsumer defaultConsumer;

    public void doConsumer(MqTopicEnum topic, MqTagEnum tag) {
        try {
            this.defaultConsumer.subscribe(topic.name(), tag.name());
            this.defaultConsumer.registerMessageListener(new MessageListenerOrderly() {
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                    msgHandle(msgs);
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });
            this.defaultConsumer.start();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    public void doBatchConsumer(MqTopicEnum topic, MqTagEnum tag, int batchSize){

    }

    /**
     * 消息处理
     * @param msgs 消息
     */
    public abstract void msgHandle(List<MessageExt> msgs);
}
